package org.caw.hbase.admin.impl;

import java.io.IOException;
import java.util.Collection;
import java.util.Map;
import java.util.TreeMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.caw.domain.descriptor.TableDescriptor;
import org.caw.util.ConvertUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class CachedHBaseAdmin extends HBaseAdmin {

	private final static Logger logger = LoggerFactory.getLogger(CachedHBaseAdmin.class);

	private Map<String, TableDescriptor> tables;

	private boolean locked;

	public CachedHBaseAdmin(Configuration conf) throws MasterNotRunningException, ZooKeeperConnectionException {
		super(conf);
		long interval = conf.getLong("hbase.cached.admin.refresh.interval", 10000);
		TableSyncThread thread = new TableSyncThread(this, interval);
		thread.start();
	}

	private synchronized void lock() {
		locked = true;
	}

	private synchronized void unlock() {
		locked = false;
	}

	private synchronized boolean isLocked() {
		return locked;
	}

	public void refresh() throws IOException {
		HTableDescriptor[] tableDescriptors = super.listTables();

		Map<String, TableDescriptor> tables = new TreeMap<String, TableDescriptor>();

		for (HTableDescriptor desc : tableDescriptors) {
			TableDescriptor tableDesc = ConvertUtil.convert(desc, true);
			// FIXME 
			//tableDesc.setEnabled(this.isTableEnabled(desc.getName()));
			tables.put(desc.getNameAsString(), tableDesc);
		}

		lock();
		this.tables = tables;
		unlock();
	}

	@Override
	public HTableDescriptor[] listTables() throws IOException {
		while (isLocked()) {
			try {
				logger.debug("Cache is locked waiting ...");
				Thread.sleep(100);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		HTableDescriptor[] desc = new HTableDescriptor[tables.values().size()];
		return tables.values().toArray(desc);
	}

	public Collection<TableDescriptor> listTablesFromCache() throws IOException {
		while (isLocked()) {
			try {
				logger.debug("Cache is locked waiting ...");
				Thread.sleep(100);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		return tables.values();
	}

	public TableDescriptor getTableDescriptor(String tableName) throws IOException {
		return tables.get(tableName);
	}

	private static class TableSyncThread extends Thread {

		private long refreshInterval;

		private CachedHBaseAdmin admin;

		public TableSyncThread(CachedHBaseAdmin admin, long interval) {
			this.refreshInterval = interval;
			this.admin = admin;
		}

		@Override
		public void run() {
			while (true) {
				try {
					logger.info("refreshing admin cache ...");
					admin.refresh();
					logger.info("admin cache is up to date");
					Thread.sleep(refreshInterval);
				} catch (InterruptedException e) {
					e.printStackTrace();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}
}
